package dds

import (
	"awesomeTask/application/client"
	"awesomeTask/application/variable"
	"awesomeTask/application/vote"
	"awesomeTask/commom/ListHelper"
	"awesomeTask/commom/MapHelper"
	"awesomeTask/commom/MessageHelper"
	"awesomeTask/commom/timeHelper"
	"awesomeTask/config"
	"awesomeTask/system/helper"
	"awesomeTask/system/kernel"
	"awesomeTask/system/log"
	"awesomeTask/system/rpc"
	"awesomeTask/system/safe"
	"fmt"
	"strconv"
	"sync"
	"unsafe"
)

const LOCAL_MAP_NAME = "LOCAL_DDS_STORE_MAP"

type DdsManager interface {
	Push(data []interface{}, key string) error                                                   //将一个数据数据集进行切分的存储在集群中
	GetChunk(key string, batch int) (err error, data interface{}, startIndex int, batchSize int) //获取一个迭代对象将将以流的方式获取数据
	GetAll(key string) (err error, data interface{})                                             //获取全部的数据
	Delete(key string)                                                                           //删除数据集
	GetDdscatalog(key string)                                                                    //获取数据集存储在集群中的位置
	Persistent(key string, data []interface{})                                                   //持久化数据存储
}

type DdsManagerObj struct {
	MaxBatchSize        int              //每个分片的最大存储数量  超过此数量将会进行自动分片
	ddsMap              sync.Map         //dds 数据存储的地址 {total:230100,slice:[{"key":"1","num":3000,"nodeid":["c0:3e:ba:10:23:d6:9980"]}],lastRequestTime:1625461505} //每个dds的数据分片将会保存在集群的任意的两个分片当中
	ddsMemory           sync.Map         //存储在内存中的dds数据
	currentMemory       int              //当前存储在存储的dds占用的内存
	CopyNum             int              //dds 副本数据的数量
	WaitPersistenceChan chan interface{} //等待持久化的数据的通道
	DdsPath             string           //持久化的dds存放的路径
}

func (dds *DdsManagerObj) randNode() []client.ClientHandle {
	connectionInfo := make([]client.ClientHandle, 0)
	safe.Lock()
	for _, node := range client.Nodes {
		if vote.HasConnected(node.UniqueId) && node.IsInit {
			connectionInfo = append(connectionInfo, node)
		}
	}
	safe.Unlock()
	res := make([]client.ClientHandle, 0)
	for i := 0; i < dds.CopyNum; i++ {
		if len(connectionInfo) == 0 {
			return res
		}
		index := helper.RandInt(0, len(connectionInfo))
		res = append(res, connectionInfo[index])
		connectionInfo = append(connectionInfo[:index], connectionInfo[index+1:]...)
	}
	return res
}

/*
持久化数据
*/
func (dds *DdsManagerObj) persistence(key string, data string) {
	datasource := map[string]interface{}{
		"key":  key,
		"data": data,
	}
	dds.WaitPersistenceChan <- datasource
}
func (dds *DdsManagerObj) recordMap(bigkey string, uniqueId string, nodeId string, count int) {
	safe.Lock()
	ddsRecords, ok := dds.ddsMap.Load(bigkey)
	if !ok {
		ddsRecords = map[string]interface{}{
			"total":    float64(0),
			"lastUsed": timeHelper.NowInt(),
			"slice":    map[string]interface{}{},
		}
	}
	mapper := map[string]interface{}{
		"count": count,
	}
	ddsRecords.(map[string]interface{})["total"] = float64(ddsRecords.(map[string]interface{})["total"].(float64)) + float64(count)
	slice, ok := ddsRecords.(map[string]interface{})["slice"].(map[string]interface{})
	if slice[uniqueId] == nil {
		slice[uniqueId] = make(map[string]interface{})
	}
	slice[uniqueId].(map[string]interface{})[nodeId] = mapper
	ddsRecords.(map[string]interface{})["slice"] = slice
	ddsRecords.(map[string]interface{})["lastUsed"] = timeHelper.NowInt()
	dds.ddsMap.Store(bigkey, ddsRecords)
	//helper.PrintJson(MapHelper.SyncMapToMap(dds.ddsMap))
	dds.persistence(LOCAL_MAP_NAME, helper.JsonEncode(MapHelper.SyncMapToMap(dds.ddsMap)))
	safe.Unlock()
}
func (dds *DdsManagerObj) Push(data []interface{}, key string) {
	chunkData := ListHelper.ListChunk(data, dds.MaxBatchSize)
	for index, item := range chunkData {
		uniqueId := key + "_" + strconv.Itoa(index)
		eventData := map[string]interface{}{
			"data":   item,
			"key":    uniqueId,
			"ddsMap": MapHelper.SyncMapToMap(dds.ddsMap),
		}
		for len(dds.randNode()) == 0 {
			//pass
		}
		for _, node := range dds.randNode() {
			MessageHelper.SendEvent(variable.ON_DDS_SAVE, eventData, node.Connetion)
			dds.recordMap(key, uniqueId, node.UniqueId, len(item))
		}
	}
}
func (dds *DdsManagerObj) changeMemroy(size int) {
	safe.Lock()
	dds.currentMemory += size
	safe.Unlock()
}

/**
保存从中心节点发送的数据同时进行存储
*/
func (dds *DdsManagerObj) SaveDDL(data string, key string, ddsMap map[string]interface{}) {
	dds.ddsMemory.Store(key, data)
	dds.changeMemroy(int(unsafe.Sizeof(data)))
	if ddsMap != nil {
		dds.persistence(key, data)
		dds.persistence(LOCAL_MAP_NAME, helper.JsonEncode(ddsMap))
	}

}

func (dds *DdsManagerObj) GetChunk(bigKey string, batch int) []interface{} {
	keyMap, ok := dds.ddsMap.Load(bigKey)
	if !ok && variable.IsMaster {
		return make([]interface{}, 0)
	}
	for sliceKey, item := range keyMap.(map[string]interface{})["slice"].(map[string]interface{}) {
		for key, _ := range item.(map[string]interface{}) {
			if key != config.GetConfigByKey("uniqueId") {
				params := map[string]interface{}{
					"key": sliceKey,
				}
				conn := client.GetNodeByUniqueId(key)
				if conn != nil {
					res, err := rpc.RpcCall(variable.ON_DDS_ASK_DATA, params, conn, 30000)
					//if err!=nil {
					//	panic(err)
					//}\
					fmt.Println("+++++++++++++++++++++++++++++++++++++++++++")
					fmt.Println(err)
					fmt.Println(res)
				}
			}
		}
	}
	return nil

}
func (dds *DdsManagerObj) GetRecordByKey(key string) interface{} {
	if value, ok := dds.ddsMemory.Load(key); ok {
		return value
	}
	path := dds.GetPath(key)
	if ok, _ := helper.PathExists(path); ok {
		return helper.ReadFileAsString(path)
	}
	return nil
}
func GetDdsObj() *DdsManagerObj {
	instance := kernel.GetIocInstance()
	return instance.GetInstanceOrCreate("dds").(*DdsManagerObj)
}

/**
从本地恢复数据
*/
func (dds *DdsManagerObj) RecoverDataByLocal() {
	path := dds.GetPath(LOCAL_MAP_NAME)
	if ok, _ := helper.PathExists(path); ok {
		data := helper.JsonDecode(helper.ReadFileAsString(path))
		fmt.Println(data)
		for key, value := range data.(map[string]interface{}) {
			dds.ddsMap.Store(key, value)
			go dds.shouldRecover(value.(map[string]interface{}))
		}
	}
}

func (dds *DdsManagerObj) shouldRecover(keyMap map[string]interface{}) {
	currentUniqueId := config.GetConfigByKey("uniqueId")
	if int64(keyMap["lastUsed"].(float64))-timeHelper.NowInt() < 60*20 {
		for key, value := range keyMap["slice"].(map[string]interface{}) {
			for uniqueId, _ := range value.(map[string]interface{}) {
				if uniqueId == currentUniqueId {
					if helper.CheckFileIsExist(dds.GetPath(key)) == false {
						log.GetLogger().Error("dds data has been losed" + dds.GetPath(key))
						return
					}
					data := helper.ReadFileAsString(dds.GetPath(key))
					dds.SaveDDL(key, data, nil)
					log.GetLogger().Info(helper.Format("ddsManager has been reload records key{} from disk", key))
				}
			}
		}
	}
}
func (dds *DdsManagerObj) Persistence() {
	for true {
		data := <-dds.WaitPersistenceChan
		key := data.(map[string]interface{})["key"].(string)
		datasource := data.(map[string]interface{})["data"].(string)
		path := dds.GetPath(key)
		helper.WriteFile(path, datasource, 0777)
	}
}
func (dds *DdsManagerObj) GetPath(key string) string {
	return dds.DdsPath + "/" + helper.Md5(key) + ".dds"
}
